package com.atguigu.sparkstreaming.examples

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/7/15
 *
 *
 *    at most once丢数据的原因:  在输出之前，已经提交了offset，一旦出现故障，只会从上次提交的位置往后消费，造成漏消费数据。
 *
 *    解决：   ①取消自动提交offset
 *              改为手动提交
 *            ②在输出成功后，再手动提交
 *
 *
 */
object AtMostOnceDemo {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "TransformDemo", Seconds(10))

    //所有的消费者参数都可以在 ConsumerConfig
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "2203092",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "true",
      //自动提交的时间间隔，没间隔多久提交一次
      "auto.commit.interval.ms" -> "500"
    )


    val topics = Array("topicA")

    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    ds.map(record => {
      Thread.sleep(500)
      if(record.value().equals("B")){
        throw new RuntimeException("程序异常");
      }
      record.value()
    }).print(1000)

    // 启动APP
    streamingContext.start()

    // 阻塞进程，让进程一直运行
    streamingContext.awaitTermination()

  }

}
